package priv.mill.test

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row


object Demo1TableApi {

  def main(args: Array[String]): Unit = {
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink的计划器
      .inStreamingMode() //使用流处理模型
      .build()

    //创建table的环境
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    //构建一个流
    val linesDS: DataStream[String] = bsEnv.socketTextStream("localhost",8888)

    //将流转换成动态表 可以指定字段
    //这个table类似于spark中的dataframe，可以使用dsl的api，flink中的dsl不是很友好，不使用
    val table: Table = bsTableEnv.fromDataStream(linesDS,$"word")

    //注册一张表
    bsTableEnv.createTemporaryView("words",table)

    //在动态表上进行连续查询
    val countTable: Table = bsTableEnv.sqlQuery(
      """
        |select word,count(1) from words group by word
      """.stripMargin)

    //更新流
    //将结果表转换成流
    //多了一个boolean，为true就是插入，为false就是删除
    val resultDS: DataStream[(Boolean, Row)] = countTable.toRetractStream[Row]

    resultDS.print()

    bsEnv.execute()

  }
}
